package juc.blockqueue.aynchronousqueue;

/**
 * @ClassName ManualThreadPool
 * @PackageName juc.blockqueue.aynchronousqueue
 * @projectName myThread
 * @Description 2. 手动线程池实现
 * @Author games
 * @Date 2025/9/4 下午1:47
 * @Version 1.0
 */

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 手动实现的线程池类，使用SynchronousQueue作为任务队列
 * 特点：
 * 1. 使用同步队列，没有容量，每个插入操作必须等待另一个线程的对应移除操作
 * 2. 线程数量动态增长，但不超过最大线程数限制
 * 3. 适用于任务量不确定且需要快速响应的场景
 */
public class ManualThreadPool {
    // 同步任务队列，用于存储待执行的任务
    private final SynchronousQueue<Runnable> taskQueue = new SynchronousQueue<>();
    // 原子整数，用于跟踪当前活跃线程数量

    /**
     * AtomicInteger 简介
     * 位于 java.util.concurrent.atomic 包中
     * 提供原子性的整数操作，无需使用 synchronized 关键字
     * 基于 CAS（Compare-And-Swap）机制实现
     *
     *
     * incrementAndGet()
     * 先增加再返回
     * 增加后的值
     * ++i
     * getAndIncrement()
     * 先返回再增加
     * 增加前的值
     * i++
     * decrementAndGet()
     * 先减少再返回
     * 减少后的值
     * --i
     * getAndDecrement()
     * 先返回再减少
     * 减少前的值
     * i--
     */
    private final AtomicInteger threadCount = new AtomicInteger(0);
    // 线程池最大线程数
    private final int maxThreads;

    /**
     * 构造函数，初始化线程池
     *
     * @param maxThreads 线程池最大线程数
     */
    public ManualThreadPool(int maxThreads) {
        this.maxThreads = maxThreads;
    }

    /**
     * 提交任务到线程池
     *
     * @param task 要执行的任务
     */
    public void submit(Runnable task) {
        // 尝试将任务交给现有线程
        if (!taskQueue.offer(task)) {
            // 如果没有空闲线程且未达到最大线程数，创建新线程
            if (threadCount.get() < maxThreads) {
                createWorker();
            }
            try {
                // 阻塞直到有线程可以处理任务
                taskQueue.put(task);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 创建工作线程
     * 内部类实现，用于从队列中获取任务并执行
     */
    private void createWorker() {
        if (threadCount.incrementAndGet() <= maxThreads) {
            new Thread(() -> {
                try {
                    while (true) {
                        Runnable task = taskQueue.take(); // 阻塞获取任务
                        System.out.println("threadCount====: " + threadCount.get());
                        task.run(); // 执行任务
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    threadCount.decrementAndGet(); // 线程结束时减少计数
                    System.out.println("threadCount----: " + threadCount.get());
                }
            }).start();
        } else {
            System.out.println("threadCount: " + threadCount.get());
            threadCount.decrementAndGet(); // 线程结束时减少计数
        }
    }

    /**
     * 主函数，测试线程池功能
     *
     * @param args 命令行参数
     * @throws InterruptedException 可能被中断的异常
     */
    public static void main(String[] args) throws InterruptedException {
        // 创建最大线程数为3的线程池
        ManualThreadPool pool = new ManualThreadPool(3);

        // 提交任务
        for (int i = 0; i < 5; i++) {
            final int taskId = i;
            pool.submit(() -> {
                System.out.println("执行任务 " + taskId + " by " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.println("任务 " + taskId + " 完成");
            });
        }
        ThreadSafeCounter counter = new ThreadSafeCounter();
        Thread.sleep(10000); // 等待任务完成
        System.out.println("计数器结果: " + counter.getValue());
    }
}

class ThreadSafeCounter {
    private AtomicInteger count = new AtomicInteger(0);

    public void increment() {
        count.incrementAndGet();
    }

    public void decrement() {
        count.decrementAndGet();
    }

    public int getValue() {
        return count.get();
    }
}
